package demo.db.zookeeper.curator;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Objects;
import java.util.SortedSet;
import java.util.TreeSet;

import com.alibaba.fastjson.JSON;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CreateBuilder;
import org.apache.curator.framework.api.ExistsBuilder;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

/**
 * <h1>节点类型</h1>
 * <li>PERSISTENT：持久化
 * <li>PERSISTENT_SEQUENTIAL：持久化并且带序列号
 * <li>EPHEMERAL：临时
 * <li>EPHEMERAL_SEQUENTIAL：临时并且带序列号
 */
public class CuratorDemo {

    public static String zookeeperAddress = "localhost:2181";

    public static CuratorFramework curator = buildCuratorFramework(zookeeperAddress, 1000, 60_000, 1000, 5, null);

    @Test
    public void testCreatePath() {
        String path = "/abc";
        try {
            createIfNotExists(path);
            String a = curator.create()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
                    .forPath(path + "/", "hello".getBytes(StandardCharsets.UTF_8));
            System.out.println(a);
        } catch (Exception e) {
            System.out.println("ZK 异常");
            e.printStackTrace();
        }

    }

    @Test
    public void testGetChildren() {
        String path = "/abc";
        try {
            GetChildrenBuilder getChildrenBuilder = curator.getChildren();
            List<String> nodes = getChildrenBuilder.forPath(path);
            System.out.println(nodes);
            SortedSet<String> sortedSet = new TreeSet<String>(nodes);
            System.err.println(sortedSet);
        } catch (Exception e) {
            System.out.println("ZK 异常");
            e.printStackTrace();
        }

    }


    /**
     * 保证锁根节点存在，若不存在则创建它
     */
    public void createIfNotExists(String path) throws Exception {
        byte[] data = "1234".getBytes(StandardCharsets.UTF_8);
        ExistsBuilder existsBuilder = curator.checkExists();
        Stat stat = existsBuilder.forPath(path);
        System.err.println(JSON.toJSONString(stat, true));
        if (stat == null) {
            CreateBuilder createBuilder = curator.create();
            String result = createBuilder.forPath(path, data);
            System.err.println(result);
        }
    }

    /**
     * @param connectString
     * @param connectionTimeoutMs
     * @param sessionTimeoutMs
     * @param baseSleepTimeMs
     * @param maxRetries
     * @param namespace
     * @return
     */
    public static CuratorFramework buildCuratorFramework(String connectString, int connectionTimeoutMs,
                                                         int sessionTimeoutMs, int baseSleepTimeMs, int maxRetries, String namespace) {
        CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                // zkClint连接地址
                .connectString(connectString)
                // 连接超时时间
                .connectionTimeoutMs(connectionTimeoutMs)
                // 会话超时时间
                .sessionTimeoutMs(sessionTimeoutMs)
                // 重试策略
                .retryPolicy(new ExponentialBackoffRetry(baseSleepTimeMs, maxRetries))
                // 命名空间,默认节点
                .namespace(namespace).build();

        curatorFramework.start();
        return curatorFramework;
    }

    public static ZooKeeper getInstance(CuratorFramework curatorFramework) throws Exception {
        if (curatorFramework == null || CuratorFrameworkState.STARTED != curatorFramework.getState()) {
            return null;
        }
        return curatorFramework.getZookeeperClient().getZooKeeper();
    }

    @Test
    public void curatorRecipesDemo() throws Exception {
        CuratorFramework curatorFramework = buildCuratorFramework(zookeeperAddress, 2000, 10000, 1000, 3,
                "myZookeeperTest");
        curatorFramework.create().forPath("/path");// 默认持久化节点,以斜杠开头
        System.out.println(curatorFramework.getChildren().forPath("/"));
        curatorFramework.create().withMode(CreateMode.EPHEMERAL).forPath("/secondPath", "hello,word".getBytes());
        System.out.println("节点secondPath的数据" + new String(curatorFramework.getData().forPath("/secondPath")));
        curatorFramework.setData().forPath("/secondPath", "hello,myWorld!".getBytes());
        System.out.println("节点secondPath的数据" + new String(curatorFramework.getData().forPath("/secondPath")));

        curatorFramework.create().creatingParentContainersIfNeeded().forPath("/secondPath/second2/second3");// 递归创建
        List<String> list = curatorFramework.getChildren().forPath("/secondPath");// 查询节点的所有字节点
        System.out.println(list);
        curatorFramework.delete().deletingChildrenIfNeeded().forPath("/secondPath/second2");// 递归删除
        System.out.println(curatorFramework.checkExists().forPath("/secondPath/second2"));// 判断节点是否存在
        System.out.println(curatorFramework.checkExists().forPath("/secondPath/second2/second3"));// 判断节点是否存在
        System.out.println(curatorFramework.getChildren().forPath("/secondPath"));
        curatorFramework.delete().deletingChildrenIfNeeded().forPath("/secondPath");

        // todo guaranteed()如果删除失败,会记录下来,只要会话有效,就会不断的重试,直到删除成功为止
        // todo Stat stat 对象包含版本id,事物id等信息
        curatorFramework.close();
    }

    @Test
    public void watcherDemo() throws Exception {
        CuratorFramework curatorFramework = buildCuratorFramework(zookeeperAddress, 2000, 10000, 1000, 3,
                "myZookeeperTest");

        curatorFramework.start();
        List<String> list = curatorFramework.getChildren().forPath("/");
        if (Objects.nonNull(list)) {
            if (!list.contains("myWatch")) {
                curatorFramework.delete().deletingChildrenIfNeeded().forPath("/myWatch");
            }
        } else {
            curatorFramework.create().forPath("/myWatch");
        }

        PathChildrenCache pathChildrenCache = pathChildrenCache = new PathChildrenCache(curatorFramework, "/myWatch",
                false);
        PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent)
                    throws Exception {
                System.out.println("pathChildrenCacheListener::::->" + pathChildrenCacheEvent.getData());
            }
        };
        pathChildrenCache.getListenable().addListener(pathChildrenCacheListener);// 注册监听事件
        pathChildrenCache.start();

        NodeCache nodeCache = new NodeCache(curatorFramework, "/myWatch", false);
        NodeCacheListener nodeCacheListener = new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                System.out.println("nodeCacheListener::::->:" + nodeCache.getCurrentData().getPath());
            }
        };
        nodeCache.getListenable().addListener(nodeCacheListener);
        nodeCache.start();

        TreeCache treeCache = new TreeCache(curatorFramework, "/myWatch");
        TreeCacheListener treeCacheListener = new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
                System.out.println("treeCacheListener::::->" + treeCacheEvent.getData());
            }
        };
        treeCache.getListenable().addListener(treeCacheListener);
        treeCache.start();

        curatorFramework.create().forPath("/myWatch/child22", "生个好孩子".getBytes());
        curatorFramework.create().creatingParentContainersIfNeeded().forPath("/myWatch/child22/child22",
                "生个好孩子".getBytes());
        curatorFramework.setData().forPath("/myWatch/child222", "生个好孩子aaaa".getBytes());
        System.in.read();// 阻塞不然启动后clint就关掉了
    }
}
